package com.freez.spark.core.demo;

import com.freez.spark.tool.TupleUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;
import scala.Tuple3;
import scala.Tuple7;

import java.util.ArrayList;
import java.util.List;

/**
 * FREEDOM  2021 人生苦短，不妨一试
 * FileName: RecommendMovie.java
 * Author: zcs
 * Date: 2021年-12月-07日 周二 18:23
 * Description: 电影推荐
 */
public class RecommendMovie {
    public static void main(String[] args) {
        String appName = "SparkDemo-recommendMovie";
        String master = "local";
        String path = "./src/main/resources/data/core/demo/movie.txt";
        recommendMovie(appName, master, path);
    }
    /**
     * 电影推荐
     *
     * @param appName Application名字
     * @param master  Spark程序运行模式
     * @param path    源文件路径
     */
    public static void recommendMovie(String appName, String master, String path) {
        SparkConf conf = new SparkConf();
        conf.setAppName(appName).setMaster(master);
        JavaSparkContext jsc = new JavaSparkContext(conf);
        JavaRDD<String> txtRDD = jsc.textFile(path);
        //(movie,user,rate) ==> (movie,(user,rate)）
        JavaPairRDD<String, Tuple2<String, Integer>> movieRDD = txtRDD.mapToPair(line -> {
            String[] split = line.split(",");
            String user = split[0];
            String movie = split[1];
            Integer rating = Integer.parseInt(split[2]);
            Tuple2<String, Integer> userRatings = TupleUtils.T2(user, rating);
            return TupleUtils.T2(movie, userRatings);
        });
        //(movie,(user,rate)） ==> (movie1,((user1,rate1),(user2,rate2)...))
        JavaPairRDD<String, Iterable<Tuple2<String, Integer>>> groupByKey = movieRDD.groupByKey();
        //(movie1,((user1,rate1),(user2,rate2)...)) ==>  (user,(movie,rating,numberOfRatingToMovie))
        JavaPairRDD<String, Tuple3<String, Integer, Integer>> groupByKeyRDD = groupByKey.flatMapToPair(s -> {
            List<Tuple2<String, Integer>> userAndRatingsList = new ArrayList<>();
            String movie = s._1;
            int numOfRaters = 0;
            Iterable<Tuple2<String, Integer>> userAndRatingsPairs = s._2;
            for (Tuple2<String, Integer> item : userAndRatingsPairs) {
                //计算每部电影评分总人数
                numOfRaters++;
                userAndRatingsList.add(item);
            }
            List<Tuple2<String, Tuple3<String, Integer, Integer>>> resultList = new ArrayList<>();
            for (Tuple2<String, Integer> item : userAndRatingsList) {
                String user = item._1;
                Integer rating = item._2;
                Tuple3<String, Integer, Integer> tuple3 = TupleUtils.T3(movie, rating, numOfRaters);
                resultList.add(TupleUtils.T2(user, tuple3));
            }
            return resultList.iterator();
        });
        //self-join
        JavaPairRDD<String, Tuple2<Tuple3<String, Integer, Integer>, Tuple3<String, Integer, Integer>>> selfJoinRDD = groupByKeyRDD.join(groupByKeyRDD);
        //去掉重复的movie对
        JavaPairRDD<String, Tuple2<Tuple3<String, Integer, Integer>, Tuple3<String, Integer, Integer>>> filterRDD = selfJoinRDD.filter(s -> {
            Tuple3<String, Integer, Integer> movie1 = s._2._1;
            Tuple3<String, Integer, Integer> movie2 = s._2._2;
            String movieName1 = movie1._1();
            String movieName2 = movie2._1();
            if (movieName1.compareTo(movieName2) < 0) {
                return true;
            }
            return false;
        });
        //（movie1-movie2,(Integer, Integer, Integer, Integer, Integer, Integer, Integer)）
        JavaPairRDD<String, Tuple7<Integer, Integer, Integer, Integer, Integer, Integer, Integer>> moviePairRDD = filterRDD.mapToPair(s -> {
            Tuple3<String, Integer, Integer> movie1 = s._2._1;
            Tuple3<String, Integer, Integer> movie2 = s._2._2;
            String moviePairName = movie1._1() + "-" + movie2._1();
            int ratingProduct = movie1._2() * movie2._2();
            int rating1Squared = movie1._2() * movie1._2();
            int rating2Squared = movie2._2() * movie2._2();
            Tuple7<Integer, Integer, Integer, Integer, Integer, Integer, Integer> t7 = TupleUtils.T7(
                    movie1._2(), //movie1评分
                    movie1._3(), //movie1评分人数
                    movie2._2(), //movie2评分
                    movie2._3(), //movie2评分人数
                    ratingProduct, //movie1.rating * movie2.rating
                    rating1Squared, //movie1.rating**2
                    rating2Squared //movie2.rating**2
            );
            return TupleUtils.T2(moviePairName, t7);
        });
        //分组排序
        JavaPairRDD<String, Iterable<Tuple7<Integer, Integer, Integer, Integer, Integer, Integer, Integer>>> resultRDD = moviePairRDD.groupByKey().sortByKey();
        //操作value值,计算相关性系数
        JavaPairRDD<String, Tuple3<Double, Double, Double>> calculateResultRDD = resultRDD.mapValues(s -> calculateCorrelation(s));
//        calculateResultRDD.foreach(s -> System.out.println(s._1 + "：" + "pearson = " + s._2._1() + "；" + "cosine = " + s._2._2() + "；" + "jaccard = " + s._2._3() + "；"));
        //打印最终结果
        for (Tuple2<String, Tuple3<Double, Double, Double>> s : calculateResultRDD.collect()) {
            System.out.println(s._1 + "：" + "pearson = " + s._2._1() + "；" + "cosine = " + s._2._2() + "；" + "jaccard = " + s._2._3() + "；");
        }
        jsc.close();
    }
    /**
     * 计算相关性系数
     *
     * @param values 一个包含了计算相关性系数的Tuple7
     * @return 返回皮尔逊积矩相关系数，余弦相似系数，杰卡德相似系数的Tuple3
     */
    public static Tuple3<Double, Double, Double> calculateCorrelation(Iterable<Tuple7<Integer, Integer, Integer, Integer, Integer, Integer, Integer>> values) {
        int groupSize = 0; // 每个vector长度
        int dotProduct = 0; // movie1和movie2评分点积和
        int rating1Sum = 0; // movie1评分和
        int rating2Sum = 0; // movie2评分和
        int rating1NormSq = 0; // movie1评分平方和
        int rating2NormSq = 0; // movie2评分平方和
        int maxNumOfRaters1 = 0;  // movie1的最大评分人数
        int maxNumOfRaters2 = 0;  // movie2的最大评分人数
        for (Tuple7<Integer, Integer, Integer, Integer, Integer, Integer, Integer> t7 : values) {
            groupSize++;
            dotProduct += t7._5();
            rating1Sum += t7._1();
            rating2Sum += t7._3();
            rating1NormSq += t7._6();
            rating2NormSq += t7._7();
            int numOfRaters1 = t7._2(); // movie1评分人数
            if (numOfRaters1 > maxNumOfRaters1) {
                maxNumOfRaters1 = numOfRaters1;
            }
            int numOfRaters2 = t7._4(); //movie2评分人数
            if (numOfRaters2 > maxNumOfRaters2) {
                maxNumOfRaters2 = numOfRaters2;
            }
        }
        double pearson = calculatePearsonCorrelation(groupSize, dotProduct, rating1Sum, rating2Sum, rating1NormSq, rating2NormSq);
        double cosine = calculateCosineCorrelation(dotProduct, Math.sqrt(rating1NormSq), Math.sqrt(rating2NormSq));
        double jaccard = calculateJaccardCorrelation(groupSize, maxNumOfRaters1, maxNumOfRaters2);
        return TupleUtils.T3(pearson, cosine, jaccard);
    }
    /**
     * 计算尔逊积矩相关系数
     *
     * @param size          double
     * @param dotProduct    double
     * @param rating1Sum    double
     * @param rating2Sum    double
     * @param rating1NormSq double
     * @param rating2NormSq double
     * @return double
     */
    private static double calculatePearsonCorrelation(double size, double dotProduct, double rating1Sum, double rating2Sum, double rating1NormSq, double rating2NormSq) {
        //协方差
        double covariance = size * dotProduct - rating1Sum * rating2Sum;
        //标准差
        double standardDeviation = Math.sqrt(size * rating1NormSq - rating1Sum * rating1Sum) * Math.sqrt(size * rating2NormSq - rating2Sum * rating2Sum);
        return covariance / standardDeviation;
    }
    /**
     * 计算余弦相似系数
     *
     * @param dotProduct
     * @param sqrt
     * @param sqrt1
     * @return
     */
    private static double calculateCosineCorrelation(double dotProduct, double sqrt, double sqrt1) {
        return dotProduct / sqrt * sqrt1;
    }
    /**
     * 计算杰卡德相似系数
     *
     * @param inCommon
     * @param totalA
     * @param totalB
     * @return
     */
    private static double calculateJaccardCorrelation(double inCommon, double totalA, double totalB) {
        double denominator = totalA + totalB - inCommon;
        return inCommon / denominator;
    }
}